package com.we.flink.userlabelplatform.table;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TableHfqAuditHomePageLog {
    public static final String KFKTABLE = "fdm.hhl01_hfq_audit_homepage_log";
    public static final String QUERYSQL =
            "select uid, user_status, created_at from "
                    + KFKTABLE
                    + " where created_at is not null";

    // created_at is not null
    public long uid;
    public String user_status;
    public int created_at;

    public static Logger LOG = LoggerFactory.getLogger(TableHfqAuditHomePageLog.class);

    public TableHfqAuditHomePageLog() {}

    public TableHfqAuditHomePageLog(long uid, String user_status, int created_at) {
        this.uid = uid;
        this.user_status = user_status;
        this.created_at = created_at;
    }

    public static TableHfqAuditHomePageLog init(long uid) {
        return new TableHfqAuditHomePageLog(uid, "#init#", 0);
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableHfqAuditHomePageLog.QUERYSQL);
        DataStream<Tuple2<Boolean, TableHfqAuditHomePageLog>> inputStream =
                tableEnv.toRetractStream(tableRes, TableHfqAuditHomePageLog.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableHfqAuditHomePageLog>>() {
                                    @Override
                                    public boolean filter(
                                            Tuple2<Boolean, TableHfqAuditHomePageLog> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TagHfqAuditHomePagelog-filter")
                        .keyBy(
                                new KeySelector<Tuple2<Boolean, TableHfqAuditHomePageLog>, Long>() {
                                    @Override
                                    public Long getKey(
                                            Tuple2<Boolean, TableHfqAuditHomePageLog> value)
                                            throws Exception {
                                        return value.f1.getUid();
                                    }
                                })
                        .flatMap(
                                new RichFlatMapFunction<
                                        Tuple2<Boolean, TableHfqAuditHomePageLog>, String>() {
                                    ValueState<TableHfqAuditHomePageLog>
                                            tableHfqAuditHomePageLogVState;

                                    @Override
                                    public void open(Configuration parameters) throws Exception {
                                        tableHfqAuditHomePageLogVState =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<
                                                                        TableHfqAuditHomePageLog>(
                                                                        "TableHfqAuditHomePageLogVState",
                                                                        TableHfqAuditHomePageLog
                                                                                .class));
                                    }

                                    @Override
                                    public void flatMap(
                                            Tuple2<Boolean, TableHfqAuditHomePageLog> value,
                                            Collector<String> out)
                                            throws Exception {
                                        long uid = value.f1.getUid();
                                        String user_status = value.f1.user_status;
                                        int created_at = value.f1.getCreated_at();

                                        TableHfqAuditHomePageLog tmpState =
                                                tableHfqAuditHomePageLogVState.value();
                                        TableHfqAuditHomePageLog curState =
                                                (tmpState == null
                                                        ? TableHfqAuditHomePageLog.init(uid)
                                                        : tmpState);

                                        if (created_at > curState.getCreated_at()
                                                && !curState.getUser_status().equals(user_status)) {
                                            JSONObject jsonObject = new JSONObject();
                                            jsonObject.put("uid", uid);
                                            jsonObject.put("user_status", user_status);
                                            out.collect(jsonObject.toString());

                                            curState.user_status = user_status;
                                            curState.created_at = created_at;
                                            tableHfqAuditHomePageLogVState.update(curState);
                                        }
                                    }
                                });

        return out;
    }

    public long getUid() {
        return uid;
    }

    public String getUser_status() {
        return user_status;
    }

    public int getCreated_at() {
        return created_at;
    }
}
